package realtime.func;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import realtime.common.GmallConfig;
import realtime.util.DruidDSUtil;

import java.sql.PreparedStatement;
import java.util.Collection;

import java.util.Set;

/**
 * @author MengX
 * @create 2023/3/24 16:46:37
 */
public class DimPhenixSinkFunction extends RichSinkFunction<JSONObject> {

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        //获取链接
        DruidPooledConnection phoenixConn = DruidDSUtil.getPhoenixConn();
        String sql = getInsertSql(value);
        System.out.println(sql);
        PreparedStatement preparedStatement = phoenixConn.prepareStatement(sql);
        preparedStatement.execute();
        phoenixConn.close();
    }

    private String getInsertSql(JSONObject value) {
        //获取插入数据的SQL语句:upsert into db.tn(id,name,sex) values('1001','zs','male')
        StringBuilder insertSql = new StringBuilder();
        String jsonStr = value.toString();
        String replace = jsonStr.replace("'", "`");
        value = JSON.parseObject(replace);
        insertSql.append("upsert into ").append(GmallConfig.HBASE_SCHEMA)
                .append(".").append(value.getString("sink_table")).append("(");
        value.remove("sink_table");
        Set<String> keys = value.keySet();
        Collection<Object> values = value.values();
        String flid = StringUtils.join(keys, ",");
        String val = StringUtils.join(values, "','");
        insertSql.append(flid).append(") values('").append(val).append("')");
        String sql = insertSql.toString();
        String[] split = sql.split("'\"'");
        if (split.length == 1){
            return sql;
        }
        return StringUtils.join(split, "'\\'");
    }
}
